package com.sub.spark.core.rdd.extend;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;

/**
 * @ClassName RDDPersist
 * @Description: RDD 持久化
 * RDD通过Cache或者Persist方法将前面的计算结果缓存，
 * 默认情况下会把数据以序列化的形式缓存在JVM的堆内存中。
 * 但是并不是这两个方法被调用时立即缓存，而是触发后面的action算子时，
 * 该RDD将会被缓存在计算节点的内存中，并供后面重用。
 * @Author Submerge.
 * @Since 2025/5/22 13:04
 * @Version 1.0
 */
public class RDDPersist {

    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf();
        sparkConf.setMaster("local[*]");
        sparkConf.setAppName("sparkCore-persist");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        List<Integer> dataList = Arrays.asList(5, 3, 1, 6, 4, 2);


        JavaRDD<Integer> javaRDD = jsc.parallelize(dataList)
                .map(x -> {
                    System.out.println("*********");
                    return x * 2;
                }).filter(x -> x % 2 == 0);
        JavaRDD<Integer> integerJavaRDD = javaRDD.flatMap((FlatMapFunction<Integer, Integer>) integer -> Collections.singletonList(integer * 2).iterator());

        //cache ：从当前RDD开始，后续的所有算子操作，都会使用当前RDD缓存的数据，
        javaRDD.cache();
        //persis：可以持久化到内存或磁盘，且StorageLevel.MEMORY_ONLY()相当于cache
        //javaRDD.persist(StorageLevel.MEMORY_ONLY());
        //javaRDD.persist(StorageLevel.MEMORY_AND_DISK());

        javaRDD.groupBy(d -> d).collect();

        System.out.println("========");

        integerJavaRDD.map(d -> "again -> " + d).collect();

        //reduceByKey，自带缓存

    }


    public static void cache() {


    }

    public static void persist() {


    }


    /**
     * 检查点：将RDD中间结果写入磁盘或HDFS
     *
     * @param sc
     */
    public static void checkPoint(JavaSparkContext sc) {
        JavaRDD<String> lineRDD = sc.textFile("data/demo/spark/num.txt");

        JavaPairRDD<String, Long> tupleRDD = lineRDD.mapToPair((PairFunction<String, String, Long>) s -> new Tuple2<>(s, System.currentTimeMillis()));

        // 查看血缘关系
        System.out.println(tupleRDD.toDebugString());

        // 增加检查点避免计算两次
        tupleRDD.cache();

        // 进行检查点
        tupleRDD.checkpoint();

        tupleRDD.collect().forEach(System.out::println);

        System.out.println(tupleRDD.toDebugString());
        // 第二次计算
        tupleRDD.collect().forEach(System.out::println);
        // 第三次计算
        tupleRDD.collect().forEach(System.out::println);

    }
}
